package com.example.rxjavacoredemo.rxjava;

import com.example.rxjavacoredemo.rxjava.function.Function;
import com.example.rxjavacoredemo.rxjava.function.ObservableObserveOn;
import com.example.rxjavacoredemo.rxjava.function.ObservableSubscribeOn;

/**
 * 被观察者
 */
public abstract class Observable<T> implements ObservableSource<T> {

    public static <T> Observable create(ObservableOnSubscribe<T> source) {
        return new ObservableCreator<T>(source);
    }

    public <R> Observable<R> map(Function<? super T, ? extends R> function) {
        return new ObservableMap(this, function);
    }

    public final Observable<T> subscribeOn(){
        return new ObservableSubscribeOn<>(this);
    }

    public Observable<T> observeOn() {
        return new ObservableObserveOn<T>(this);
    }

    @Override
    public void subscribe(Observer observer) {
        subscribeActual(observer);
    }

    protected abstract void subscribeActual(Observer<? super T> observer);
}
